package chapter5.eg1;

import chapter5.AbstractConsumer;
import java.time.Duration;
import java.util.Collections;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.junit.Test;

public class ConsumerTest extends AbstractConsumer {

  @Test
  public void testConsumer() {
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
    consumer.subscribe(Collections.singletonList("test-topic"));
    try {
      Duration duration = Duration.ofMillis(1000);
      while (true) {
        ConsumerRecords<String, String> records = consumer.poll(duration);
        for (ConsumerRecord<String, String> record : records) {
          printRecord(record);
        }
      }
    } finally {
      consumer.close();
    }
  }
}
